package exchange

import (
	"github.com/garyburd/redigo/redis"
	"global/libs/structs"
	"global/constants"
	"github.com/gorilla/websocket"
	"time"
	"github.com/ngaut/log"
	"strconv"
	"global/util"
	"github.com/shopspring/decimal"
	"encoding/json"
	"strings"
	"fmt"
	"global/core"
	"sync"
)

type HuobiTickerResponse struct {
	Rep 		string 		`json:"rep"`
	Status 		string 		`json:"status"`
	Id 			string 		`json:"id"`
	Data 		struct{
		Amount 			decimal.Decimal		`json:"amount"`
		Open 			decimal.Decimal		`json:"open"`
		Close			decimal.Decimal 	`json:"close"`
		High 			decimal.Decimal 	`json:"high"`
		Id 				uint64				`json:"id"`
		Count 			uint64				`json:"count"`
		Low 			decimal.Decimal		`json:"low"`
		Version 		uint64	 			`json:"version"`
		Volume 			decimal.Decimal 	`json:"volume"`
	}						`json:"data"`
}

type Huobi struct {
	Slug 				string
	Pool 				*redis.Pool
	IntervalC 			*time.Ticker
	Symbols 			[]structs.Symbol
	FormatSymbols 		map[string]structs.Symbol
	Wss 				map[string]*websocket.Conn
	mux 				sync.Mutex
}

func NewHuobiTicker(pool *redis.Pool, slug string, symbols []structs.Symbol) InterfaceTicker {
	h := &Huobi{
		Pool:		pool,
		Slug: 		slug,
		Symbols: 	symbols,
		IntervalC: 	time.NewTicker(1 * time.Second),
		FormatSymbols: make(map[string]structs.Symbol),
		Wss: 		make(map[string]*websocket.Conn),
	}
	h.formatSymbol()
	return h
}

func (e *Huobi) NeedSymbols() bool {
	return true
}

func (e *Huobi) ListenAndAccept() error {
	e.connects()
	c := time.NewTicker(time.Second * 5)
	for t := range c.C {
		_ = t
		log.Infof("[%s] websocket connects: %d", e.Slug, len(e.Wss))
	}

	return nil
}

func (e *Huobi) formatSymbol()  {
	for _, v := range e.Symbols {
		key := fmt.Sprintf("market.%s.detail", strings.ToLower(v.From + v.To))
		e.FormatSymbols[key] = v
	}
}

func (e *Huobi) connects() {
	for k, s := range e.FormatSymbols {
		go func(key string, symbol structs.Symbol) {
			for {
				ws, _, err := websocket.DefaultDialer.Dial(constants.HuobiWebsocketBaseUrl, nil)
				if err != nil {
					log.Errorf("Error: socket dial failed, k: %s, msg: %s", key, err.Error())
					time.Sleep(time.Second * 5)
					continue
				}

				e.saveConn(key, ws)

				// 每秒发送一次请求
				go func() {
					for t := range e.IntervalC.C {
						_ = t
						err := e.Request(key, key)
						if err != nil {
							log.Errorf("Error: [%s]request ticker failed, msg: %s", e.Slug, err)
						}
					}
				}()

				// 处理消息
				for {
					_, msg, err := ws.ReadMessage()
					if err != nil {
						log.Errorf("Error: read message failed, msg: %s", err)
						if websocket.IsUnexpectedCloseError(err) {
							log.Errorf("is unexcepted error")
						}
						break
					}

					go func() {
						data, err := util.UnGzip(msg)
						if err != nil {
							log.Errorf("Error: Ungzip error, msg: %s", err)
						}

						e.handlerResponse(key, data)
					}()
				}

				e.removeConn(key)
				if ws != nil {
					ws.Close()
				}
			}
		}(k, s)
	}
}

func (e *Huobi) removeConn(key string) {
	e.mux.Lock()
	if _, ok := e.Wss[key]; ok {
		delete(e.Wss, key)
	}
	e.mux.Unlock()
}

// 保存连接
func (e *Huobi) saveConn(key string, conn *websocket.Conn) {
	e.mux.Lock()
	e.Wss[key] = conn
	e.mux.Unlock()
}

func (e *Huobi) handlerResponse(key string, message []byte) {
	resp := util.JsonDecodeByte(message)
	if pong, ok := resp["ping"]; ok {
		e.mux.Lock()
		if ws, ok := e.Wss[key]; ok {
			ws.WriteJSON(map[string]interface{}{"pong": pong})
		}
		e.mux.Unlock()
		return
	}

	if _, ok := resp["rep"]; ok {
		resp, err := e.responseParse(message)
		if err != nil {
			log.Errorf("Error: %s", err)
			return
		}

		log.Infof("get message[%s], %s", key, string(message))

		symbol, ok := e.FormatSymbols[resp.Rep]
		if !ok {
			log.Errorf("Error: symbol not found, symbol=%s", resp.Rep)
			return
		}

		tickers := make([]structs.Ticker, 0)
		tickers = append(tickers, structs.Ticker{
			Symbol: 		symbol.From + "_" + symbol.To,
			From: 			symbol.From,
			To: 			symbol.To,
			Volume: 		resp.Data.Volume.StringFixed(8),
			Amount: 		resp.Data.Amount.StringFixed(8),
			Percent: 		"",
			Open: 			resp.Data.Open.StringFixed(8),
			High: 			resp.Data.High.StringFixed(8),
			Low: 			resp.Data.Low.StringFixed(8),
			Close: 			resp.Data.Close.StringFixed(8),
			Number: 		resp.Data.Count,
			PriceCny: 		"",
			PriceUsd: 		"",
		})

		err = core.SaveTicker(e.Pool, e.Slug, tickers)
		if err != nil {
			log.Errorf("Error: save tickers failed, msg: %s", err)
		}
	}
}

func (e *Huobi)responseParse(data []byte) (*HuobiTickerResponse, error) {
	var resp HuobiTickerResponse
	err := json.Unmarshal(data, &resp)
	if err != nil {
		return nil, err
	}
	return &resp, nil
}

func (e *Huobi) Request(key string, req string) error {
	var message structs.ReqStruct
	message.Id = strconv.Itoa(time.Now().Nanosecond())
	message.Req = req

	e.mux.Lock()
	defer e.mux.Unlock()

	if ws, ok := e.Wss[key]; ok {
		return ws.WriteJSON(message)
	}
	return nil
}




